﻿#pragma once

#include <hacks/base/hacks_conf.h>
#include <hacks/base/ev_utils.h>
#include <hacks/net/io_buff.h>

namespace hacks ::net {

    class tcp_conn;

    typedef std::shared_ptr<tcp_conn> shared_tcp_conn;

    typedef std::function<void(tcp_conn &, char *data, std::size_t len)> conn_data_functor_t;
    typedef std::function<void(tcp_conn &)> conn_self_functor_t;
    typedef conn_self_functor_t conn_new_functor_t;
    typedef conn_self_functor_t conn_closed_functor_t;

    class tcp_conn : public std::enable_shared_from_this<tcp_conn> {

    public:
        tcp_conn(const shared_ev_loop &loop,
                 ev_os_sock_t fd);

        void async_write(const_buffer &&data);

        void commit();

        void set_conn_id(const std::string &s);

        const std::string &get_conn_id() const;

        void set_accept_callback(conn_new_functor_t &&f);

        void set_data_callback(conn_data_functor_t &&f);

        void set_closed_callback(conn_closed_functor_t &&f);

    private:
        friend class tcp_server;

        void on_accepted();

        void conn_rw_handler(ev_sock_fd &s, int status, int event) {
            if (status < 0) {
                s.close();
                return;
            }
            bool drop_read = data_functor_ == nullptr;
            const auto &loop = s.get_loop();
            auto &loop_buf = loop->get_loop_buffer();
            auto os_fd = s.get_os_fd();
            if (event & ev_event_readable) {
                for (;;) {
                    ssize_t recv_len =
                            ::recv(os_fd, loop_buf.data, loop_buf.len, 0/* no flag*/);
                    if (recv_len == 0) {
                        // todo on_closed eof !!!
                        s.close();
                        return;
                    }
                    //drop read data
                    if (drop_read) { continue; }
                    if (recv_len > 0) {
                        //on data read
                        if (data_functor_) { data_functor_(*this, loop_buf.data, static_cast<std::size_t> (recv_len)); }
                        continue;
                    }
                    if (recv_len == -1 && errno == EAGAIN) {
                        s.update_event(s.get_current_event() | ev_event_writeable);
                        break;
                    }
                    if (recv_len == -1 && errno != EAGAIN) {
                        //fatal error
                        s.close();
                        return;
                    }
                    break;
                }
            }

            if (event & ev_event_writeable) {
                if (s.get_status() != evt_commited) {
                    return;
                }
                for (;;) {
                    if (const_buffers_.empty()) { break; }
                    auto &item_const_buf = const_buffers_.front();
                    const auto *buffer_data = item_const_buf.buff.data() + item_const_buf.w_idx;
                    const auto buff_data_len = item_const_buf.buff.size() - item_const_buf.w_idx;
                    ssize_t send_len = ::send(os_fd,
                                              buffer_data,
                                              buff_data_len,
                                              OPT_NO_SIG_PIPE);
                    if (send_len != -1 && send_len != 0) {
                        //change cursor
                        item_const_buf.w_idx += buff_data_len;
                        item_const_buf.on_progress(item_const_buf.w_idx);
                    }
                    if (item_const_buf.buff.size() == item_const_buf.w_idx) {
                        const_buffers_.pop_front();
                        item_const_buf.on_complete();
                        continue;
                    }
                    if (send_len == -1) {
                        if (errno == EAGAIN) {
                            //try again
                            s.update_event(s.get_current_event() | ev_event_writeable);
                        } else {
                            s.close();
                        }
                    } else {
                        //more to write
                        s.update_event(s.get_current_event() | ev_event_writeable);
                    }
                    break;
                }
            }
        }

        void conn_closed_handler(ev_sock_fd &s) {
            if (closed_functor_) {
                closed_functor_(*this);
            }
        }

    private:
        shared_ev_loop loop_;
        ev_os_sock_t sock_fd_;
        shared_ev_sock_fd shd_sock_fd_;
        const_buffer_list_t const_buffers_;
        //
        std::once_flag once_commit_;
        //
        std::string conn_id_;
        //callbacks
        conn_new_functor_t accept_functor_;
        conn_data_functor_t data_functor_;
        conn_closed_functor_t closed_functor_;
    };

}
